package top.wintp.offlinedataanalysis.etl.mr;

import com.google.gson.Gson;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.zip.CRC32;

import top.wintp.offlinedataanalysis.common.ConstantValues;
import top.wintp.offlinedataanalysis.common.EventLogConstant;
import top.wintp.offlinedataanalysis.etl.utils.LogUtils;


/**
 * description: 读取原始的log文件，筛选字段，获取可用数据 入HBase库
 * <p>
 * author:  upuptop
 * <p>
 * qq: 337081267
 * <p>
 * CSDN:   http://blog.csdn.net/pyfysf
 * <p>
 * cnblogs:   http://www.cnblogs.com/upuptop
 * <p>
 * blog:   http://wintp.top
 * <p>
 * email:  pyfysf@163.com
 * <p>
 * time: 2019/08/2019/8/13
 * <p>
 */
public class AnalyserETLMapper extends Mapper<LongWritable, Text, NullWritable, Put> {
    private static final Logger logger = LoggerFactory.getLogger(AnalyserETLMapper.class);
    private CRC32 crc32 = new CRC32();

    /**
     * 成功数据的计数器
     */
    private Counter successCounter;
    /**
     * 失败个数的计数器
     */
    private Counter failCounter;
    /**
     * 列族名称
     */
    private String familyName = EventLogConstant.EVENT_LOGS_FAMILY_NAME;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //获取计数器
        successCounter = context.getCounter(ConstantValues.SUCCESS_COUNTER_TAG, "SUCCESS");
        failCounter = context.getCounter(ConstantValues.FAIL_COUNTER_TAG, "FAIL");
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //158.179.38.72^A1465562718.152^A59410^A/log.gif?ver=1&pl=website&sdk=js&l=zh-CN&b_rst=1536*864&c_time=1465562718000&en=e_pv&u_ud=1dfdd8f3-f18e-4147-802a-b74ecfafa1f1&u_sd=dd9b9b17-afc9-4cd6-97e9-f851c2e7bc1b&u_mid=dd9b9b17-afc9-4cd6-97e9-f851c2e7bc1b&b_iev=Mozilla%2F5.0+%28Windows+NT+6.3%3B+WOW64%3B+rv%3A45.0%29+Gecko%2F20100101+Firefox%2F45.0&p_url=http%3A%2F%2Flocalhost%3A8080%2FBIG_DATA_LOG2%2Fdemo124.jsp&p_ref=http%3A%2F%2Flocalhost%3A8080%2FBIG_DATA_LOG2%2Fdemo145.jsp
        String line = value.toString();

        if (StringUtils.isEmpty(line)) {
            //    数据为空
            logFail();
            return;
        }

        try {
            //检查日志
            Map<String, String> paramMap = LogUtils.checkLog(line);

            //    解析日志
            LogUtils.parserLog(paramMap);


            String resultJson = new Gson().toJson(paramMap);
            logger.info("AnalyserETLMapper  map()   resultJson   " + resultJson);

                /*
                    设计rowkey,时间戳+用户id+en事件名称  最后crc32
                */
            String uuid = paramMap.get(EventLogConstant.LOG_COLUMN_NAME_UUID);
            String memberId = paramMap.get(EventLogConstant.LOG_COLUMN_NAME_MEMBER_ID);
            String serverTime = paramMap.get(EventLogConstant.LOG_COLUMN_NAME_SERVER_TIME);

            if (StringUtils.isEmpty(serverTime)) {
                throw new Exception("service_time 为空");
            }

            String eventAliasName = EventLogConstant.EventEnum.valueOfAlias(paramMap.get(EventLogConstant.LOG_COLUMN_NAME_EVENT)).alias;
            String rowKey = genRowkey(serverTime, uuid, memberId, eventAliasName);

            //将map里面的数据都封装到put里面
            Put put = new Put(Bytes.toBytes(rowKey));

            for (Map.Entry<String, String> param : paramMap.entrySet()) {
                if (StringUtils.isNotBlank(param.getKey()) && StringUtils.isNotBlank(param.getValue())) {
                    put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(param.getKey()), Bytes.toBytes(param.getValue()));
                }
            }

            //数据写出
            context.write(NullWritable.get(), put);

            logSuccess();
        } catch (Exception e) {
            e.printStackTrace();
            logFail();
        }

    }

    /**
     * 构建rowkey
     *
     * @param serverTime
     * @param uuid
     * @param memberId
     * @param eventAliasName
     * @return
     */
    private String genRowkey(String serverTime, String uuid, String memberId, String eventAliasName) {


        StringBuilder sb = new StringBuilder();
        sb.append(serverTime).append("_");
        this.crc32.reset();
        if (StringUtils.isNotBlank(uuid)) {
            this.crc32.update(uuid.getBytes());
        }
        if (StringUtils.isNotBlank(memberId)) {
            this.crc32.update(memberId.getBytes());
        }
        this.crc32.update(eventAliasName.getBytes());
        sb.append(this.crc32.getValue() % 100000000L);
        return sb.toString();
    }

    /**
     * 不合格的log
     */
    private void logFail() {
        failCounter.increment(1);
    }

    /**
     * 合格的log
     */
    private void logSuccess() {
        successCounter.increment(1);
    }


    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        super.cleanup(context);
        long successCounterValue = successCounter.getValue();
        long failCounterValue = failCounter.getValue();


        logger.info("成功数据{}条，失败数据{}条。      ", successCounterValue, failCounterValue);
    }
}
